package com.bw.gmall.realtime.ods;

import com.alibaba.fastjson.JSON;
import com.bw.gmall.realtime.bean.AdsDmpBasicFeature;
import com.bw.gmall.realtime.bean.DwsUserAgeFeature;
import com.bw.gmall.realtime.bean.OdsUserBasicInfo;
import com.bw.gmall.realtime.bean.OdsUserBehavior;
import lombok.Data;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * 达摩盘用户画像基础特征分层计算Flink Job
 * 工单编号：大数据-用户画像-11-达摩盘基础特征
 */
public class DmpBasicFeatureFlinkJob {

    // Kafka配置常量
    private static final String KAFKA_BOOTSTRAP_SERVERS = "kafka-node1:9092,kafka-node2:9092";
    private static final String KAFKA_GROUP_ID = "dmp_basic_feature_group";
    private static final String USER_BEHAVIOR_TOPIC = "ods_user_behavior";
    private static final String USER_BASIC_INFO_TOPIC = "ods_user_basic_info";

    // Hive输出常量
    private static final String HIVE_TABLE = "hive_catalog.dmp_user_profile.ads_dmp_basic_feature";

    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink执行环境（批流一体）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings tableEnvSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, tableEnvSettings);

        // 2. 读取Kafka数据源
        DataStream<OdsUserBehavior> userBehaviorStream = readUserBehaviorStream(env);
        DataStream<OdsUserBasicInfo> userBasicInfoStream = readUserBasicInfoStream(env);

        // 3. DWD层：数据清洗与标准化
        DataStream<DwdUserBehaviorClean> cleanedBehaviorStream = cleanUserBehavior(userBehaviorStream);
        DataStream<DwdUserBasicInfoClean> cleanedBasicInfoStream = cleanUserBasicInfo(userBasicInfoStream);

        // 4. DWS层：特征汇总计算
        SingleOutputStreamOperator<DwdUserBehaviorClean> ageFeatureStream = calculateAgeFeatures(cleanedBehaviorStream);
        SingleOutputStreamOperator<DwdUserBehaviorClean> genderFeatureStream = calculateGenderFeatures(cleanedBehaviorStream);
        SingleOutputStreamOperator<DwdUserBasicInfoClean> bodyFeatureStream = calculateBodyFeatures(cleanedBasicInfoStream);

        // 5. ADS层：最终标签生成
        DataStream<AdsDmpBasicFeature> resultStream = combineFeatures(ageFeatureStream, genderFeatureStream, bodyFeatureStream);

        // 6. 结果输出
        writeToHive(tableEnv, resultStream);

        // 执行Flink Job
        env.execute("DmpBasicFeatureFlinkJob");
    }

    private static DataStream<AdsDmpBasicFeature> combineFeatures(SingleOutputStreamOperator<DwdUserBehaviorClean> ageFeatureStream, SingleOutputStreamOperator<DwdUserBehaviorClean> genderFeatureStream, SingleOutputStreamOperator<DwdUserBasicInfoClean> bodyFeatureStream) {
        return null;
    }

    /**
     * 读取用户行为数据
     */
    private static DataStream<OdsUserBehavior> readUserBehaviorStream(StreamExecutionEnvironment env) {
        return env.addSource(new FlinkKafkaConsumer<>(
                        USER_BEHAVIOR_TOPIC,
                        new SimpleStringSchema(),
                        getKafkaProperties()))
                .map(json -> JSON.parseObject(json, OdsUserBehavior.class))
                .name("UserBehaviorSource")
                .uid("user-behavior-source");
    }

    /**
     * 读取用户基础信息数据
     */
    private static DataStream<OdsUserBasicInfo> readUserBasicInfoStream(StreamExecutionEnvironment env) {
        return env.addSource(new FlinkKafkaConsumer<>(
                        USER_BASIC_INFO_TOPIC,
                        new SimpleStringSchema(),
                        getKafkaProperties()))
                .map(json -> JSON.parseObject(json, OdsUserBasicInfo.class))
                .name("UserBasicInfoSource")
                .uid("user-basic-info-source");
    }

    /**
     * 创建Kafka配置
     */
    private static Properties getKafkaProperties() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS);
        props.setProperty("group.id", KAFKA_GROUP_ID);
        // 添加必要的消费者配置
        props.setProperty("auto.offset.reset", "latest");
        props.setProperty("enable.auto.commit", "false");
        return props;
    }

    /**
     * 清洗用户行为数据
     */
    private static DataStream<DwdUserBehaviorClean> cleanUserBehavior(DataStream<OdsUserBehavior> sourceStream) {
        return sourceStream
                .map(new DwdUserBehaviorCleanMap())
                .name("UserBehaviorCleaner")
                .uid("user-behavior-cleaner");
    }

    /**
     * 清洗用户基础信息数据
     */
    private static DataStream<DwdUserBasicInfoClean> cleanUserBasicInfo(DataStream<OdsUserBasicInfo> sourceStream) {
        return sourceStream
                .map(new DwdUserBasicInfoCleanMap())
                .name("UserBasicInfoCleaner")
                .uid("user-basic-info-cleaner");
    }

    /**
     * 计算年龄特征
     */
    private static SingleOutputStreamOperator<DwdUserBehaviorClean> calculateAgeFeatures(DataStream<DwdUserBehaviorClean> cleanedStream) {
        return cleanedStream
                .keyBy(DwdUserBehaviorClean::getUserId)
                .window(TumblingProcessingTimeWindows.of(Time.of(1, TimeUnit.HOURS)))
                .reduce(new DwsUserAgeFeatureReduce())
                .name("AgeFeatureCalculator")
                .uid("age-feature-calculator");
    }

    /**
     * 计算性别特征
     */
    private static SingleOutputStreamOperator<DwdUserBehaviorClean> calculateGenderFeatures(DataStream<DwdUserBehaviorClean> cleanedStream) {
        return cleanedStream
                .keyBy(DwdUserBehaviorClean::getUserId)
                .window(TumblingProcessingTimeWindows.of(Time.of(1, TimeUnit.HOURS)))
                .reduce(new DwsUserGenderFeatureReduce())
                .name("GenderFeatureCalculator")
                .uid("gender-feature-calculator");
    }

    /**
     * 计算身体特征（身高、体重、星座）
     */
    private static SingleOutputStreamOperator<DwdUserBasicInfoClean> calculateBodyFeatures(DataStream<DwdUserBasicInfoClean> cleanedStream) {
        return cleanedStream
                .keyBy(DwdUserBasicInfoClean::getUserId)
                .window(TumblingProcessingTimeWindows.of(Time.of(1, TimeUnit.HOURS)))
                .reduce(new DwsUserBodyFeatureReduce())
                .name("BodyFeatureCalculator")
                .uid("body-feature-calculator");
    }

    /**
     * 合并多维度特征
     */
//    private static DataStream<AdsDmpBasicFeature> combineFeatures(
//            DataStream<DwdUserBehaviorClean> ageStream,
//            DataStream<DwdUserBehaviorClean> genderStream,
//            DataStream<DwdUserBasicInfoClean> bodyStream) {
//
//        // 首先连接年龄和性别流
//        ConnectedStreams<DwdUserBehaviorClean, DwdUserBehaviorClean> ageGenderConnected = ageStream.connect(genderStream);
//        DataStream<AdsDmpBasicFeature> ageGenderMerged = ageGenderConnected.map(
//                age -> {
//                    AdsDmpBasicFeature feature = new AdsDmpBasicFeature();
//                    // 设置年龄相关属性
//                    feature.setUserId(age.getUserId());
//                    // ... 其他属性设置
//                    return feature;
//                },
//                gender -> {
//                    AdsDmpBasicFeature feature = new AdsDmpBasicFeature();
//                    // 设置性别相关属性
//                    feature.setUserId(gender.getUserId());
//                    // ... 其他属性设置
//                    return feature;
//                }
//        );
//
//        // 再与身体特征流连接
//        return ageGenderMerged.connect(bodyStream)
//                .map(
//                        ads -> ads,
//                        body -> {
//                            AdsDmpBasicFeature feature = new AdsDmpBasicFeature();
//                            // 设置身体特征相关属性
//                            feature.setUserId(body.getUserId());
//                            // ... 其他属性设置
//                            return feature;
//                        }
//                )
//                .keyBy(AdsDmpBasicFeature::getUserId)
//                .reduce((f1, f2) -> {
//                    // 合并同一用户的特征
//                    AdsDmpBasicFeature merged = new AdsDmpBasicFeature();
////                    merged.setUserId(f1.getUserId());
//                    // 合并其他属性
////                    merged.setAgeSegment(f1.getAgeSegment() != null ? f1.getAgeSegment() : f2.getAgeSegment());
////                    merged.setGenderTag(f1.getGenderTag() != null ? f1.getGenderTag() : f2.getGenderTag());
//                    // ... 其他属性合并
//                    merged.setUpdateTime(new Timestamp(System.currentTimeMillis()));
//                    return merged;
//                })
//                .name("FeatureCombiner")
//                .uid("feature-combiner");
//    }

    /**
     * 写入Hive表
     */
    private static void writeToHive(StreamTableEnvironment tableEnv, DataStream<AdsDmpBasicFeature> resultStream) {
        Table resultTable = tableEnv.fromDataStream(resultStream);
        tableEnv.createTemporaryView("ads_dmp_basic_feature", resultTable);

        tableEnv.executeSql("INSERT INTO " + HIVE_TABLE + " " +
                "SELECT userId, ageSegment, genderTag, heightCm, weightKg, constellation, updateTime, dataDate " +
                "FROM ads_dmp_basic_feature");
    }

    // ------------------------------ 自定义处理逻辑实现 ------------------------------

    /**
     * DWD层用户行为清洗MapFunction：实现类目匹配、搜索词分类、设备属性判定
     */
    static class DwdUserBehaviorCleanMap implements MapFunction<OdsUserBehavior, DwdUserBehaviorClean> {
        // 类目映射表 - 改为静态常量，避免重复创建
        private static final Map<String, String> CATEGORY_MAP;

        static {
            CATEGORY_MAP = new HashMap<>();
            CATEGORY_MAP.put("cat1001", "潮流服饰");
            CATEGORY_MAP.put("cat1002", "家居用品");
            CATEGORY_MAP.put("cat1003", "健康食品");
        }

        @Override
        public DwdUserBehaviorClean map(OdsUserBehavior value) throws Exception {
            DwdUserBehaviorClean clean = new DwdUserBehaviorClean();
            clean.setUserId(value.getUserId());
            clean.setBehaviorType(value.getBehaviorType());
//            clean.setBrandName(value.getBrandName());
//            clean.setPriceRange(value.getPriceRange());
            clean.setDt(value.getDt());

            // 1. 匹配商品类目名称
            clean.setCategoryName(getCategoryName(value.getCategoryId()));

            // 2. 搜索词分类
            clean.setSearchKeywordType(getSearchKeywordType(value.getSearchKeyword()));

            // 3. 设备属性判定
            clean.setHighEndDevice(isHighEndDevice(value.getDeviceType(), value.getDeviceVersion()));
            clean.setLatestVersion(isLatestVersion(value.getDeviceVersion()));

            return clean;
        }

        private String getCategoryName(String categoryId) {
            return CATEGORY_MAP.getOrDefault(categoryId, "其他");
        }

        private String getSearchKeywordType(String keyword) {
            if (keyword == null) return "其他";
            if (keyword.matches(".*潮流|新款|时尚.*")) return "时尚与潮流";
            if (keyword.matches(".*优惠|折扣|便宜.*")) return "性价比";
            if (keyword.matches(".*健康食品|保健品|养生.*")) return "健康与养生";
            if (keyword.matches(".*婴儿用品|儿童玩具|亲子.*")) return "家庭与育儿";
            if (keyword.matches(".*智能手机|电脑配件|智能设备.*")) return "科技与数码";
            if (keyword.matches(".*在线课程|书籍|技能提升.*")) return "学习与发展";
            return "其他";
        }

        private int isHighEndDevice(String deviceType, String version) {
            if (deviceType == null || version == null) return 0;
            if ("ios".equals(deviceType) && version.matches("iOS 18.*")) return 1;
            if ("android".equals(deviceType) && version.matches("Android 15.*")) return 1;
            return 0;
        }

        private int isLatestVersion(String version) {
            if (version == null) return 0;
            return version.matches("iOS 18.*|Android 15.*") ? 1 : 0;
        }
    }

    /**
     * DWD层用户基础信息清洗MapFunction：身高体重单位统一、异常值过滤
     */
    static class DwdUserBasicInfoCleanMap implements MapFunction<OdsUserBasicInfo, DwdUserBasicInfoClean> {
        @Override
        public DwdUserBasicInfoClean map(OdsUserBasicInfo value) throws Exception {
            DwdUserBasicInfoClean clean = new DwdUserBasicInfoClean();
            clean.setUserId(value.getUserId());
            clean.setDt(value.getDt());

            // 1. 生日清洗
            clean.setBirthDate(validateBirthDate(value.getBirthDate()));

            // 2. 身高清洗
            clean.setHeightDevice(validateHeight(value.getDeviceHeight()));
            clean.setHeightAuth(validateHeight(parseHeight(value.getInputHeight())));
            clean.setHeightOrder(validateHeight(parseHeight(value.getOrderHeight())));
            clean.setHeightForm(validateHeight(parseHeight(value.getFormHeight())));

            // 3. 体重清洗
            clean.setWeightDevice(validateWeight(value.getDeviceWeight()));
            clean.setWeightAuth(validateWeight(parseWeight(value.getInputWeight())));
            clean.setWeightOrder(validateWeight(parseWeight(value.getOrderWeight())));
            clean.setWeightForm(validateWeight(parseWeight(value.getFormWeight())));

            return clean;
        }

        private String validateBirthDate(String birthDate) {
            if (birthDate != null && birthDate.matches("^\\d{4}-\\d{2}-\\d{2}$")) {
                return birthDate;
            }
            return null;
        }

        private Double parseHeight(String heightStr) {
            if (heightStr == null) return null;
            String numStr = heightStr.replaceAll("[^0-9.]", "");
            if (numStr.isEmpty()) return null;
            try {
                return Double.parseDouble(numStr);
            } catch (NumberFormatException e) {
                return null;
            }
        }

        private Double validateHeight(Double height) {
            if (height != null && height >= 100 && height <= 250) {
                return height;
            }
            return null;
        }

        private Double parseWeight(String weightStr) {
            if (weightStr == null) return null;
            String numStr = weightStr.replaceAll("[^0-9.]", "");
            if (numStr.isEmpty()) return null;
            try {
                double num = Double.parseDouble(numStr);

                if (weightStr.contains("斤")) {
                    return num / 2; // 斤转kg
                } else if (weightStr.contains("g")) {
                    return num / 1000; // g转kg
                } else {
                    return num; // 默认kg
                }
            } catch (NumberFormatException e) {
                return null;
            }
        }

        private Double validateWeight(Double weight) {
            if (weight != null && weight >= 30 && weight <= 200) {
                return weight;
            }
            return null;
        }
    }

    /**
     * DWS层年龄特征汇总ReduceFunction：按权重计算各维度得分
     */
    static class DwsUserAgeFeatureReduce implements ReduceFunction<DwdUserBehaviorClean> {
        @Override
        public DwdUserBehaviorClean reduce(DwdUserBehaviorClean value1, DwdUserBehaviorClean value2) throws Exception {
            // 1. 类目偏好得分计算（权重30%）
            Map<String, Double> categoryScore1 = calculateCategoryScore(value1.getCategoryName());
            Map<String, Double> categoryScore2 = calculateCategoryScore(value2.getCategoryName());
            Map<String, Double> categoryScore = mergeScoreMap(categoryScore1, categoryScore2, 0.3);

            // 2. 品牌偏好得分计算（权重20%）
            Map<String, Double> brandScore1 = calculateBrandScore(value1.getBrandName());
            Map<String, Double> brandScore2 = calculateBrandScore(value2.getBrandName());
            Map<String, Double> brandScore = mergeScoreMap(brandScore1, brandScore2, 0.2);

            // 3. 合并所有得分并确定最终年龄分段
            String ageSegment = determineAgeSegment(categoryScore, brandScore);

            // 设置结果
            DwdUserBehaviorClean result = new DwdUserBehaviorClean();
            result.setUserId(value1.getUserId());
            // ... 设置其他需要传递的属性
            return result;
        }

        private String determineAgeSegment(Map<String, Double> categoryScore, Map<String, Double> brandScore) {
            // 实现年龄分段判定逻辑
            double maxScore = -1;
            String ageSegment = "未知";

            for (String age : categoryScore.keySet()) {
                double totalScore = categoryScore.get(age) + brandScore.getOrDefault(age, 0.0);
                if (totalScore > maxScore) {
                    maxScore = totalScore;
                    ageSegment = age;
                }
            }
            return ageSegment;
        }

        private Map<String, Double> calculateCategoryScore(String categoryName) {
            Map<String, Double> scoreMap = new HashMap<>();
            switch (categoryName) {
                case "潮流服饰":
                    scoreMap.put("18-24", 0.9);
                    scoreMap.put("25-29", 0.8);
                    scoreMap.put("30-34", 0.6);
                    scoreMap.put("35-39", 0.4);
                    scoreMap.put("40-49", 0.2);
                    scoreMap.put("50以上", 0.1);
                    break;
                case "家居用品":
                    scoreMap.put("18-24", 0.2);
                    scoreMap.put("25-29", 0.4);
                    scoreMap.put("30-34", 0.6);
                    scoreMap.put("35-39", 0.8);
                    scoreMap.put("40-49", 0.9);
                    scoreMap.put("50以上", 0.7);
                    break;
                case "健康食品":
                    scoreMap.put("18-24", 0.1);
                    scoreMap.put("25-29", 0.2);
                    scoreMap.put("30-34", 0.4);
                    scoreMap.put("35-39", 0.6);
                    scoreMap.put("40-49", 0.8);
                    scoreMap.put("50以上", 0.9);
                    break;
                default:
                    // 初始化所有年龄分段的得分为0
                    initScoreMap(scoreMap);
            }
            return scoreMap;
        }

        private Map<String, Double> calculateBrandScore(String brandName) {
            Map<String, Double> scoreMap = new HashMap<>();
            // 实现品牌偏好得分计算逻辑
            initScoreMap(scoreMap);
            // ... 根据品牌名称设置不同年龄分段的得分
            return scoreMap;
        }

        private void initScoreMap(Map<String, Double> scoreMap) {
            scoreMap.put("18-24", 0.0);
            scoreMap.put("25-29", 0.0);
            scoreMap.put("30-34", 0.0);
            scoreMap.put("35-39", 0.0);
            scoreMap.put("40-49", 0.0);
            scoreMap.put("50以上", 0.0);
        }

        private Map<String, Double> mergeScoreMap(Map<String, Double> map1, Map<String, Double> map2, double weight) {
            Map<String, Double> mergedMap = new HashMap<>();
            for (String age : map1.keySet()) {
                double sum = map1.get(age) + map2.getOrDefault(age, 0.0);
                mergedMap.put(age, sum * weight);
            }
            return mergedMap;
        }
    }

    /**
     * DWS层性别特征汇总ReduceFunction
     */
    static class DwsUserGenderFeatureReduce implements ReduceFunction<DwdUserBehaviorClean> {
        @Override
        public DwdUserBehaviorClean reduce(DwdUserBehaviorClean value1, DwdUserBehaviorClean value2) throws Exception {
            // 实现性别特征计算逻辑
            DwdUserBehaviorClean result = new DwdUserBehaviorClean();
            result.setUserId(value1.getUserId());
            // ... 计算并设置性别相关特征
            return result;
        }
    }

    /**
     * DWS层身体特征汇总ReduceFunction
     */
    static class DwsUserBodyFeatureReduce implements ReduceFunction<DwdUserBasicInfoClean> {
        @Override
        public DwdUserBasicInfoClean reduce(DwdUserBasicInfoClean value1, DwdUserBasicInfoClean value2) throws Exception {
            // 实现身高体重加权计算和星座解析逻辑
            DwdUserBasicInfoClean result = new DwdUserBasicInfoClean();
            result.setUserId(value1.getUserId());
            // ... 计算并设置身体相关特征
            return result;
        }
    }

    // ------------------------------ 数据模型类 ------------------------------

    @Data
    static class DwdUserBehaviorClean {
        private String userId;
        private String behaviorType;
        private String categoryName;
        private String brandName;
        private String priceRange;
        private String searchKeywordType;
        private int highEndDevice;
        private int latestVersion;
        private Timestamp behaviorTime;
        private String dt;
    }

    @Data
    static class DwdUserBasicInfoClean {
        private String userId;
        private String birthDate;
        private Double heightDevice;
        private Double heightAuth;
        private Double heightOrder;
        private Double heightForm;
        private Double weightDevice;
        private Double weightAuth;
        private Double weightOrder;
        private Double weightForm;
        private Timestamp updateTime;
        private String dt;
    }

    @Data
    static class DwsUserGenderFeature {
        private String userId;
        private Double femaleScore;
        private Double maleScore;
        private Double familyPurchaseRatio;
        private String statDate;
    }

    @Data
    static class DwsUserBodyFeature {
        private String userId;
        private Double weightedHeight;
        private Double weightedWeight;
        private String constellation;
        private String statDate;
    }
}